I can review this if you're interested :)

I can review this if you're interested :)

Thanks, that would be helpful! I've updated the description :)
Although in order to proceed I think we would need approval from one maintainer too.

Hi @paualarco sorry for ignoring this

Looking at the bug example:

    KafkaConsumerObservable
      .manualCommit[String, String](kafkaConfig, List(topic))
      .timeoutOnSlowUpstreamTo(5.seconds, Observable.empty)
      .foldLeft(CommittableOffsetBatch.empty) { case (batch, message) => batch.updated(message.committableOffset) }
      .mapEval(completeBatch => completeBatch.commitAsync())
      .headOrElseL(List.empty)

I feel like the real issue is that completeBatch.commitAsync() returns before the commit is completed? IIRC the underlying consumer is not closed until the downstream completes and it feels like it should complete after the last commit is done. Is the issue reproducible with these changes #250 ?

@Avasil no problem,

I don't think that's the case, so the poll heartbeat consumer would fix the problem of partition reassignment by rebalancing, which was caused by slow downstream consumers.
However in this case the issue is different, it happens to be that when the observable is cancelled (in the example because it is reaching a certain timeout), the consumer is being closed too, but that's not desirable when there are elements that have not yet been committed. Thus, when they do, they return an error indicating that the consumer was already closed.

It gets fixed when the resource is responsible of closing the consumer, and not from the Observable onComplete.

For the poll heartbeat to have been caused this issue, the max.poll.interval.ms would have needed to be higher than 300sec (as the default value), but the test were not as high :)